package com.nulldev.util.concurrency.threadIt.v4.emapi.ft2;

import java.lang.reflect.Field;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.concurrency.threadIt.v3.extensions.Task;
import com.nulldev.util.concurrency.threadIt.v3.extensions.TaskStatus;
import com.nulldev.util.concurrency.threadIt.v4.asyncTasks.AsyncTaskCallback;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.locking.ResettableCountDownLatch;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.TaskTimedOutException;
import com.nulldev.util.internal.backport.concurrency9.Lists;
import com.nulldev.util.internal.unsafecompat.IUnsafe;
import com.nulldev.util.internal.unsafecompat.UnsafeCompat;
import com.nulldev.util.io.FastIO.FastIO;
import com.nulldev.util.manage.Version;
import com.nulldev.util.manage.legacy.Timer;

/**
 * STAGING VARIANT OF FUTURETASK <br>
 * EXPECT EXPLOSIONS AND FEATURES
 * 
 * @author null
 * @api-version 4.0
 * @since 3.0.1
 * @param <T>
 */
@SuppressWarnings("deprecation")
public abstract class FutureTask2<T> implements Runnable, Executable<T> {

	private static final IUnsafe UNSAFE = UnsafeCompat.get();
	private static final long THREAD_STATUS_OFFSET;

	static {
		final Field field;
		try {
			field = FutureTask2.class.getDeclaredField("threadStatus");
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		THREAD_STATUS_OFFSET = UNSAFE.objectFieldOffset(field);
	}

	public static final double VERSION = 1.08;
	public static final Version LIB_VERSION = new Version("1.0.8");

	private final static boolean USE_LAZY_AWAIT = Arguments.notArguments("--Tasks.disableLazyAwait");
	/*
	 * FIXME: Events should NOT be chained and should execute with priority. Disable
	 * async-event-submission for now.
	 */
	private final static boolean ENABLE_ASYNC_EVENT_SUBMISSION = Arguments.hasArgument("--Tasks.enableAsyncEventSubmission");

	private final BlockingQueue<AsyncTaskCallback<T>> callback;

	private int threadStatus = TaskStatus.UNSUBMITTED.getID();

	private final AtomicReference<T> result = new AtomicReference<T>();
	private final AtomicReference<Throwable> cachedError = new AtomicReference<Throwable>();
	private final AtomicReference<Thread> t = new AtomicReference<Thread>();
	public final Throwable e;

	private final ResettableCountDownLatch countDownLatch = new ResettableCountDownLatch(1);

	public FutureTask2() {
		this.callback = new ArrayBlockingQueue<AsyncTaskCallback<T>>(16, true);
		this.e = new Exception("/// TRACE /// [" + MathUtil.toHex(super.hashCode()) + "]").fillInStackTrace();
	}

	public FutureTask2(final BlockingQueue<AsyncTaskCallback<T>> callback) {
		this.callback = callback;
		this.e = new Exception("\\\\\\ TRACE \\\\\\  [" + MathUtil.toHex(super.hashCode()) + "]").fillInStackTrace();
	}

	@Override
	public void run(final Thread currentThread) {
		if (!UNSAFE.compareAndSwapInt(this, THREAD_STATUS_OFFSET, TaskStatus.QUEUED.getID(), TaskStatus.RUNNING.getID()))
			throw new RuntimeException("FT2: Thread status is desynchronized.");

		this.t.lazySet(currentThread);
		try {
			final T r = this.execute();
			this.result.lazySet(r);
			if (!UNSAFE.compareAndSwapInt(this, THREAD_STATUS_OFFSET, TaskStatus.RUNNING.getID(), TaskStatus.COMPLETED.getID()))
				throw new RuntimeException("FT2: Thread status is desynchronized.");
			if (!this.callback.isEmpty()) {
				this.callback.parallelStream().forEach((callback) -> {
					if (callback == null)
						return;
					ExecutorManager.get().autoAsyncSubmit(ENABLE_ASYNC_EVENT_SUBMISSION, () -> {
						try {
							callback.onCompletion(r);
						} catch (Throwable e) {
							e.printStackTrace(System.err);
							System.err.flush();
						}
					});
				});
				this.callback.clear();
			}
		} catch (Throwable e) {
			if (!UNSAFE.compareAndSwapInt(this, THREAD_STATUS_OFFSET, TaskStatus.RUNNING.getID(), TaskStatus.FAILED.getID()))
				throw new RuntimeException("FT2: Thread status is desynchronized.");
			this.cachedError.lazySet(e);
			if (!this.callback.isEmpty()) {
				this.callback.parallelStream().forEach((callback) -> {
					if (callback == null)
						return;
					ExecutorManager.get().autoAsyncSubmit(ENABLE_ASYNC_EVENT_SUBMISSION, () -> {
						try {
							callback.onError(e);
						} catch (Throwable ex) {
							ex.printStackTrace(System.err);
							System.err.flush();
						}
					});
				});
				this.callback.clear();
			} else {
				FastIO.get().printlnError(e);
			}
		} finally {
			this.countDownLatch.countDown();
		}
	}

	@Override
	public void run() {
		this.run(null);
	}

	public abstract T execute();

	/**
	 * Freezes the current thread until either the following occur: <br>
	 * - The task completes and returns the result. <br>
	 * - The task fails, is killed and <code>NULL</code> is returned.
	 */
	public T waitTillCompletion() {
		if (this.countDownLatch.isCompleted() || this.completed()) {
			if (this.threadStatus == TaskStatus.FAILED.getID())
				throw new RuntimeException(this.cachedError.get());
			else if (this.threadStatus == TaskStatus.SUBMISSION_FAILED.getID())
				throw new RuntimeException("Submission was rejected.");
			else
				return this.result.get();
		} else {
			try {
				if (USE_LAZY_AWAIT)
					this.countDownLatch.lazyAwait();
				else
					this.countDownLatch.await();
			} catch (InterruptedException e) {
			}
			if (this.threadStatus == TaskStatus.FAILED.getID())
				throw new RuntimeException(this.cachedError.get());
			else if (this.threadStatus == TaskStatus.SUBMISSION_FAILED.getID())
				throw new RuntimeException("Submission was rejected.");
			else
				return this.result.get();
		}
	}

	@Override
	public boolean completed() {
		return this.threadStatus == TaskStatus.COMPLETED.getID() || this.threadStatus == TaskStatus.FAILED.getID()
				|| this.threadStatus == TaskStatus.KILLED.getID() || this.threadStatus == TaskStatus.CANCELLED.getID();
	}

	@Override
	public TaskStatus status() {
		return TaskStatus.from(this.threadStatus);
	}

	@Override
	public T result() {
		return this.result.get();
	}

	/**
	 * Attempts to kill the executing thread.
	 */
	public void kill() {
		UNSAFE.putInt(this, THREAD_STATUS_OFFSET, TaskStatus.KILLED.getID());
		if (this.t.get() != null) {
			this.t.get().interrupt();
		}
		try {
			this.countDownLatch.countDown();
		} catch (Exception e) {
		}
	}

	public boolean addCallbackTask(final AsyncTaskCallback<T> task) {
		if (task == null) {
			return false;
		} else if (this.countDownLatch.isCompleted() || this.completed()) {

			if (this.threadStatus == TaskStatus.FAILED.getID()) {
				task.onError(this.cachedError.get());
			} else if (this.threadStatus == TaskStatus.SUBMISSION_FAILED.getID()) {
				task.onSubmissionRejection();
			} else {
				try {
					task.onCompletion(this.result.get());
				} catch (Throwable tx) {
					task.onError(tx);
				}
			}
			return true;

		} else {
			try {
				return this.callback.offer(task);
			} catch (Throwable _internal) {
				if (this.completed()) {
					if (this.threadStatus == TaskStatus.FAILED.getID()) {
						task.onError(this.cachedError.get());
					} else if (this.threadStatus == TaskStatus.SUBMISSION_FAILED.getID()) {
						task.onSubmissionRejection();
					} else {
						task.onCompletion(this.result.get());
					}
					return true;
				} else {
					FastIO.get().printlnError(_internal);
					return false;
				}
			}
		}
	}

	@Deprecated
	public Task<T> toLegacyTask() {
		return new Task<T>(this.callback) {

			@Override
			public T run() {
				FutureTask2.this.run();
				return FutureTask2.this.result.get();
			}

		};
	}

	@Override
	public List<AsyncTaskCallback<?>> callbacks() {
		return Lists.of(this.callback.toArray(new AsyncTaskCallback[this.callback.size()]));
	}

	@Override
	public void signalQueued() {
		if (!UNSAFE.compareAndSwapInt(this, THREAD_STATUS_OFFSET, TaskStatus.UNSUBMITTED.getID(), TaskStatus.QUEUED.getID()))
			throw new RuntimeException("FT2: Thread status is desynchronized.");
		if (!this.callback.isEmpty()) {
			this.callback.parallelStream().forEach((_o) -> {
				if (_o == null)
					return;
				final AsyncTaskCallback<T> o = _o;
				ExecutorManager.get().autoAsyncSubmit(ENABLE_ASYNC_EVENT_SUBMISSION, () -> {
					try {
						o.onQueued();
					} catch (Throwable ec) {
						ec.printStackTrace(System.err);
						System.err.flush();
					}
				});
			});
		}
	}

	@Override
	public void rejectSubmission() {
		UNSAFE.putInt(this, THREAD_STATUS_OFFSET, TaskStatus.SUBMISSION_FAILED.getID());
		if (!this.callback.isEmpty()) {
			/* FIXME: Should we use offloaded-async here? */
			this.callback.iterator().forEachRemaining((o) -> {
				if (o == null)
					return;
				try {
					o.onSubmissionRejection();
				} catch (Throwable ec) {
					ec.printStackTrace(System.err);
					System.err.flush();
				}
			});
		}
	}

	@Override
	public T waitFor(final long amount, final TimeUnit unit) {
		if (amount < 0 || unit == null)
			throw new IllegalArgumentException("amount < 0!");
		if (this.countDownLatch.isCompleted() || this.completed())
			if (this.threadStatus == TaskStatus.FAILED.getID())
				throw new RuntimeException(this.cachedError.get());
			else if (this.threadStatus == TaskStatus.SUBMISSION_FAILED.getID())
				throw new RuntimeException("Submission was rejected.");
			else
				return this.result.get();
		else {
			try {
				if (this.countDownLatch.await(amount, unit)) {
					if (this.t.get() != null)
						this.t.get().interrupt();
					throw new TaskTimedOutException(toString() + ": Timed out at waitFor(" + amount + ", " + unit + ")!", this.e);
				}
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
			if (this.threadStatus == TaskStatus.FAILED.getID())
				throw new RuntimeException(this.cachedError.get());
			else if (this.threadStatus == TaskStatus.SUBMISSION_FAILED.getID())
				throw new RuntimeException("Submission was rejected.");
			else
				return this.result.get();
		}
	}

	@Override
	public T run_st() {
		this.run();
		return this.result.get();
	}

	@Override
	public Throwable getException() {
		return this.cachedError.get();
	}

	@Override
	public String toString() {
		return "FutureTask2[staging=true,hash=" + MathUtil.toHex(super.hashCode()) + ",state=" + TaskStatus.read(this.threadStatus) + ",result=" + this.result
				+ "]";
	}

	@Override
	public boolean isRunning() {
		return this.threadStatus == TaskStatus.RUNNING.getID();
	}

	/* FT 1.1 */

	@Override
	public boolean requestCancel() {
		if (this.t.get() == null)
			return false;
		try {
			this.t.get().notifyAll();
			this.t.get().interrupt();
			return true;
		} catch (Throwable ex) {
			return false;
		}
	}

	@Override
	public Object getFlag(final int flag) {
		switch (flag) {
			case FT_SUPPORTS_TRACES:
			case FT_SUPPORTS_POLLING:
				return FT_TRUE;
			default:
				return FT_NOT_SUPPORTED;
		}
	}

	@Override
	public int getFlags() {
		return FT_SUPPORTS_TRACES | FT_SUPPORTS_POLLING;
	}

	@Override
	public boolean requestReset() {
		this.t.lazySet(null);
		UNSAFE.putInt(this, THREAD_STATUS_OFFSET, TaskStatus.UNSUBMITTED.getID());
		this.result.lazySet(null);
		this.cachedError.lazySet(null);
		this.callback.clear();
		try {
			this.countDownLatch.countDownThenReset();
			return true;
		} catch (Exception ex) {
			return false;
		}
	}

	@Override
	public Throwable getTrace() {
		return this.e;
	}

	@Override
	public boolean hasTrace() {
		return this.e != null;
	}

	@Override
	public boolean hasExecutingThread() {
		return this.t.get() != null;
	}

	@Override
	public Thread getExecutingThread() {
		return this.t.get();
	}

	/* FT 1.1: Polling */

	@Override
	public T pollUntilCompletion(final int polling_algorithm) {
		switch (polling_algorithm) {
			case FT_POLLING_IMMEDIATE: {
				while (this.isRunning()) {
					try {
						Thread.sleep(1L);
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
				}
				break;
			}
			case FT_POLLING_NEW_SCALING: {
				// TODO: Implement and create an algorithm for FT_POLLING_NEW_SCALING.
			}
			default:
			case FT_POLLING_LEGACY_SCALING: {
				final int limit = 8;
				int waitingInt = limit;
				final int s = (int) (FT_POLLING_LEGACY_SCALING_WAITING_INTERVAL / limit);
				while (this.isRunning()) {
					if (waitingInt <= 0) {
						waitingInt = limit + 1;
					}
					try {
						Thread.sleep((long) Math.ceil(s / waitingInt));
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
					waitingInt -= 1;
				}
				break;
			}
		}
		if (this.threadStatus == TaskStatus.FAILED.getID())
			throw new RuntimeException(this.cachedError.get());
		else
			return this.result.get();
	}

	@Override
	public T pollFor(final long amount, final TimeUnit unit, final int polling_algorithm) {
		final Timer time = new Timer();
		time.start(ExecutorManager.get().executor());
		final long am = unit.toMillis(amount);
		switch (polling_algorithm) {
			case FT_POLLING_IMMEDIATE: {
				while (this.isRunning() && !time.hasReached(am)) {
					try {
						Thread.sleep(1L);
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
				}
				break;
			}
			case FT_POLLING_NEW_SCALING: {
				// TODO: Implement and create an algorithm for FT_POLLING_NEW_SCALING.
			}
			default:
			case FT_POLLING_LEGACY_SCALING: {
				final int limit = 8;
				int waitingInt = limit;
				final int s = (int) (FT_POLLING_LEGACY_SCALING_WAITING_INTERVAL / limit);
				while (this.isRunning() && !time.hasReached(am)) {
					if (waitingInt <= 0) {
						waitingInt = limit + 1;
					}
					try {
						Thread.sleep((long) Math.ceil(s / waitingInt));
					} catch (InterruptedException e) {
						Thread.interrupted();
					}
					waitingInt -= 1;
				}
				break;
			}
		}
		time.stop();
		if (this.threadStatus == TaskStatus.FAILED.getID())
			throw new RuntimeException(this.cachedError.get());
		else
			return this.result.get();
	}
}
